package com.flinksql.test;

import com.flinksql.bean.WaterSensor;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * @author: Lin
 * @create: 2021-06-16 10:21
 * @description: FlinkSQL，查询未注册表
 **/
public class FlinkSQL_Test1 {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        DataStreamSource<String> source = env.socketTextStream("hadoop102", 9999);
        SingleOutputStreamOperator<WaterSensor> mapDS = source.map(new RichMapFunction<String, WaterSensor>() {
            @Override
            public WaterSensor map(String value) throws Exception {
                String[] split = value.split(",");
                return new WaterSensor(split[0]
                        , Long.parseLong(split[1])
                        , Integer.parseInt(split[2]));
            }
        });

        EnvironmentSettings setting = EnvironmentSettings
                .newInstance()
                .inStreamingMode()
                .build();

        //1.创建表的执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env,setting);
        //2.创建表：将流转换成动态表。
        Table table = tableEnv.fromDataStream(mapDS);
        //3.使用sql查询未注册表
        Table sqlQuery = tableEnv.sqlQuery("select * from " + table);

        tableEnv.toAppendStream(sqlQuery, Row.class).print();


        env.execute();
    }
}
